package org.apache.lucene.ocean;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.ocean.Batch.MasterBatch;
import org.apache.lucene.ocean.Index.IndexSnapshot;
import org.apache.lucene.ocean.WriteableMemoryIndex.MemoryIndexSnapshot;
import org.apache.lucene.store.RAMDirectory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Abstract transaction class. Right now it probably implements too much
 * functionality but I'm not quite sure yet. The call method needs to be
 * overridden to implement the core functionality.
 * 
 * The deletes from the batch are applied to each IndexSnapshot.
 * AddWriteableMemoryDocumentsTask is executed on the existing
 * WriteableMemoryIndex to create a snapshot and IndexReader with the new
 * documents.
 * 
 */
public abstract class AbstractTransaction extends Transaction implements Callable<CommitResult> {
  final static Logger LOG = LoggerFactory.getLogger(AbstractTransaction.class);
  protected Batch batch;
  /**
   * Collection consisting of the failures
   */
  protected List<Failure> failures = new ArrayList<Failure>();
  /**
   * 
   */
  protected List<IndexSnapshotCommitResult> deletesResults = new ArrayList<IndexSnapshotCommitResult>();
  protected List<IndexSnapshot> newIndexSnapshots = new ArrayList<IndexSnapshot>();
  /**
   * Snapshot id/transaction id
   */
  protected Long id;
  protected TransactionSystem system;
  protected Snapshot previousSnapshot;
  protected int numDocsAdded = 0;
  /**
   * Currently not used
   */
  // TODO: figure out what this is for
  protected final List<Long> documentIds;

  public AbstractTransaction(Long id, List<Long> documentIds, Snapshot previousSnapshot, Batch batch, TransactionSystem system)
      throws Exception {
    this.id = id;
    this.documentIds = documentIds;
    this.previousSnapshot = previousSnapshot;
    this.batch = batch;
    this.system = system;
  }

  public Snapshot getPreviousSnapshot() {
    return previousSnapshot;
  }
  
  /**
   * Get the transaction id
   */
  public Long getId() {
    return id;
  }

  /**
   * Creates the newSnapshot and returns a CommitResult if there were no errors.
   * If there were errors then rollback the snapshot on each Index.
   * 
   * @return
   * @throws Exception
   */
  protected CommitResult createCommitResult() throws Exception {
    if (failures.size() == 0) {
      List<IndexSnapshot> indexSnapshots = new ArrayList<IndexSnapshot>(previousSnapshot.getDeleteOnlyIndexSnapshots());
      indexSnapshots.addAll(newIndexSnapshots);
      WriteableMemoryIndex writeableIndex = (WriteableMemoryIndex) previousSnapshot.getWriteableSnapshot().getIndex();
      Snapshot newSnapshot = new Snapshot(new BigDecimal(id), writeableIndex.getLatestIndexSnapshot(), indexSnapshots, system, System
          .currentTimeMillis());
      for (IndexSnapshotCommitResult deletesResult : deletesResults) {
        // TODO: maybe change deletesresult as the getnumadded method makes it
        // somewhat
        // misleading
        numDocsAdded += deletesResult.getNumAdded();
      }
      CommitResult commitResult = new CommitResult(newSnapshot, documentIds, deletesResults, numDocsAdded, writeableIndex.getId());
      return commitResult;
    } else {
      // rollback indexes
      LOG.info("rolling back snapshot: " + id);
      for (IndexSnapshot indexSnapshot : previousSnapshot.getIndexSnapshots()) {
        indexSnapshot.getIndex().rollback(id);
      }
      // TODO: need assertion that rollbacks worked
      throw new Exception("transaction failed " + failures);
    }
  }
  
  /**
   * For multithreaded transactions, when all of them 
   * are ready, then this method will return true.  Otherwise
   * it will return false.
   */
  public boolean go() {
    return true;
  }
  
  /**
   * When a call is 
   */
  public void ready(Index index) {
  }

  /**
   * If the batch is a MasterBatch then record information from it and write it
   * to the transaction log.
   */
  protected void handleMasterBatch() {
    try {
      if (batch instanceof MasterBatch) {
        MasterBatch masterBatch = (MasterBatch) batch;
        if (masterBatch.hasDeletes()) {
          int numDocIds = 0;
          for (IndexSnapshotCommitResult deletesResult : deletesResults) {
            numDocIds += deletesResult.getDocIds().size();
          }
          long[] docIds = new long[numDocIds];
          int x = 0;
          for (IndexSnapshotCommitResult deletesResult : deletesResults) {
            for (Long docId : deletesResult.getDocIds()) {
              docIds[x] = docId;
              x++;
            }
          }
          masterBatch.getDeletes().setDocIds(docIds);
        }
        system.getTransactionLog().writeMasterBatch(id, masterBatch);
      }
    } catch (Exception exception) {
      failures.add(new LogFailure(exception));
    }
  }

  /**
   * If the transaction fails for an index this method records it
   */
  public void failed(Index index, Throwable throwable) {
    failures.add(new IndexFailure(index, throwable));
  }

  /**
   * From a RAMDirectory create a new RamIndex.  If the number of documents 
   * in the RAMDirectory is less than a certain number then a MemoryIndexSnapshot
   * is created because this removes the need for a merge later on.
   * 
   */
  public class AddRamIndexDocumentsTask implements Callable<IndexSnapshotCommitResult> {
    private RAMDirectory ramDirectory;
    private WriteableMemoryIndex writeableIndex;

    public AddRamIndexDocumentsTask(RAMDirectory ramDirectory, WriteableMemoryIndex writeableIndex) {
      this.ramDirectory = ramDirectory;
      this.writeableIndex = writeableIndex;
    }

    public IndexSnapshotCommitResult call() throws Exception {
      // TODO: create new ramindex
      IndexReader indexReader = IndexReader.open(ramDirectory);
      if (indexReader.maxDoc() > 30) {
        long indexIdNum = system.getNextRamIndexId();
        IndexID indexId = new IndexID(indexIdNum, "ram");
        Analyzer analyzer = batch.getAnalyzer();
        RamIndex ramIndex = new RamIndex(indexId, id, null, ramDirectory, system);
        IndexSnapshot indexSnapshot = ramIndex.commitIndex(AbstractTransaction.this);
        newIndexSnapshots.add(indexSnapshot);
        IndexSnapshotCommitResult deletesResult = new IndexSnapshotCommitResult(indexId);
        addDeletesResult(deletesResult);
        return deletesResult;
      } else {
        MemoryIndexSnapshot latestIndexSnapshot = writeableIndex.getLatestIndexSnapshot();
        writeableIndex.createIndexSnapshot(id, latestIndexSnapshot.cloneReaders(true));
        IndexSnapshotCommitResult deletesResult = new IndexSnapshotCommitResult(writeableIndex.getId());
        addDeletesResult(deletesResult);
        return deletesResult;
      }
    }
  }
  
  /**
   * Given the Documents, calls commitChanges on the current WriteableMemoryIndex
   */
  public class AddWriteableMemoryDocumentsTask implements Callable<IndexSnapshotCommitResult> {
    private Documents documents;
    private Analyzer analyzer;
    private Deletes deletes;
    private WriteableMemoryIndex writeableIndex;

    public AddWriteableMemoryDocumentsTask(Documents documents, Analyzer analyzer, Deletes deletes, WriteableMemoryIndex writeableIndex) {
      this.documents = documents;
      this.analyzer = analyzer;
      this.deletes = deletes;
      this.writeableIndex = writeableIndex;
    }

    public IndexSnapshotCommitResult call() throws Exception {
      IndexSnapshotCommitResult deletesResult = writeableIndex.commitChanges(documents, deletes, analyzer, AbstractTransaction.this);
      addDeletesResult(deletesResult);
      return deletesResult;
    }
  }

  void addDeletesResult(IndexSnapshotCommitResult deletesResult) {
    assert deletesResult != null;
    deletesResults.add(deletesResult);
  }
  
  /**
   * For the given Index commit the Deletes
   *
   */
  public class DeletesTask implements Callable<IndexSnapshotCommitResult> {
    private Index index;
    private Deletes deletes;

    public DeletesTask(Deletes deletes, Index index) {
      this.deletes = deletes;
      this.index = index;
    }

    public IndexSnapshotCommitResult call() throws Exception {
      IndexSnapshotCommitResult deletesResult = index.commitDeletes(deletes, AbstractTransaction.this);
      newIndexSnapshots.add(index.getLatestIndexSnapshot());
      addDeletesResult(deletesResult);
      return deletesResult;
    }
  }
  
  /**
   * Abstract class for Failures (errors) during the transaction process
   *
   */
  public abstract static class Failure extends Exception {
    private String string;

    public Failure(String message) {
      super(message);
      string = message;
    }

    public Failure(Throwable throwable) {
      super(throwable);
      string = ExceptionUtils.getFullStackTrace(throwable);
    }

    public String toString() {
      return string;
    }
  }
  
  /**
   * Index failure
   */
  public static class IndexFailure extends Failure {
    Index index;

    public IndexFailure(Index index, Throwable throwable) {
      super(throwable);
      this.index = index;
    }
  }
  
  /**
   * Timeout failure
   */
  public static class TimeoutFailure extends Failure {
    public TimeoutFailure(String message) {
      super(message);
    }
  }
  
  /**
   * Transaction Log failure
   */
  public static class LogFailure extends Failure {
    public LogFailure(Throwable throwable) {
      super(throwable);
    }
  }
  
  /**
   * Get the new IndexSnapshots created by this transaction
   * @return
   */
  public List<IndexSnapshot> getNewIndexSnapshots() {
    return newIndexSnapshots;
  }
}
